package cn.doitedu.rtmk.demo8;

import cn.doitedu.rtmk.common.EventBean;
import groovy.lang.GroovyClassLoader;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

@Slf4j
public class EngineProcessFunction extends KeyedBroadcastProcessFunction<Long, EventBean, RuleMetaBean, String> {


    // 运算机池
    final HashMap<String, RuleCalculator> calculatorHashMap = new HashMap<>();

    GroovyClassLoader groovyClassLoader;


    MapState<String, RuleCalculator> calculatorStatePool;

    @Override
    public void open(Configuration parameters) throws Exception {
        groovyClassLoader = new GroovyClassLoader();

        calculatorStatePool = getRuntimeContext().getMapState(new MapStateDescriptor<String, RuleCalculator>("calculator_pool", String.class, RuleCalculator.class));

    }

    @Override
    public void processElement(EventBean eventBean, KeyedBroadcastProcessFunction<Long, EventBean, RuleMetaBean, String>.ReadOnlyContext readOnlyContext, Collector<String> collector) throws Exception {

        for (Map.Entry<String, RuleCalculator> entry : calculatorHashMap.entrySet()) {
            RuleCalculator ruleCalculator = entry.getValue();
            ruleCalculator.calc(eventBean,collector);

        }

//        for (Map.Entry<String, RuleCalculator> entry : calculatorStatePool.entries()) {
//            RuleCalculator ruleCalculator = entry.getValue();
//            ruleCalculator.calc(eventBean,collector);
//        }

//        MapStateDescriptor<String, RuleCalculator> bcStateDesc = new MapStateDescriptor<>("bc_state", String.class, RuleCalculator.class);
//        ReadOnlyBroadcastState<String, RuleCalculator> broadcastStateCalculatorPool = readOnlyContext.getBroadcastState(bcStateDesc);
//
//        for (Map.Entry<String, RuleCalculator> entry : broadcastStateCalculatorPool.immutableEntries()) {
//            RuleCalculator ruleCalculator = entry.getValue();
//            ruleCalculator.calc(eventBean, collector);
//        }


    }

    @Override
    public void processBroadcastElement(RuleMetaBean ruleMetaBean, KeyedBroadcastProcessFunction<Long, EventBean, RuleMetaBean, String>.Context context, Collector<String> collector) throws Exception {
        MapStateDescriptor<String, RuleCalculator> bcStateDesc = new MapStateDescriptor<>("bc_state", String.class, RuleCalculator.class);
        BroadcastState<String, RuleCalculator> broadcastStateCalculatorPool = context.getBroadcastState(bcStateDesc);

        // 收到的 ruleMetaBean ，有可能代表 管理平台 增加、修改、删除了规则参数

        // 如果操作是新增或更新后，则我们的逻辑相同，都是构造运算机，并放入运算机池
        if (("+I".equals(ruleMetaBean.getOp()) || "+U".equals(ruleMetaBean.getOp())) && ruleMetaBean.getOnlineStatus() == 1) {

            // 构造运行机，并使用传入的规则参数来初始化该运算机对象
            // 原来上直接new：  RuleCalculator ruleCalculator = new RuleModel_1_Calculator();

            // 现在，要从注入的代码中去构造

            Class aClass = groovyClassLoader.parseClass(ruleMetaBean.getRuleModelCode());
            RuleCalculator ruleCalculator = (RuleCalculator) aClass.newInstance();


            // 初始化时，向运算机对象，注入： 规则id，规则参数json，规则的目标预圈选人群bitmap
            ruleCalculator.init(getRuntimeContext(), ruleMetaBean.getRuleParamJson(), ruleMetaBean.getPreSelectUsersBitmap());

            // 将初始化好的运算机，放入运算机池
            calculatorHashMap.put(ruleMetaBean.getRuleId(),ruleCalculator);
            //calculatorStatePool.put(ruleMetaBean.getRuleId(),ruleCalculator);  // 行不通，广播数据处理方法中没有key上下文信息
            //broadcastStateCalculatorPool.put(ruleMetaBean.getRuleId(), ruleCalculator);


            log.warn("收到上线规则的请求，成功上线了规则运算机: {}", ruleMetaBean.getRuleId());

        }
        // 如果操作上 删除，则我们需要把该规则从运算机池移除
        else if ("-D".equals(ruleMetaBean.getOp()) || ("+U".equals(ruleMetaBean.getOp()) && ruleMetaBean.getOnlineStatus() == 0)) {
            calculatorHashMap.remove(ruleMetaBean.getRuleId());
            // calculatorStatePool.remove(ruleMetaBean.getRuleId());
            //broadcastStateCalculatorPool.remove(ruleMetaBean.getRuleId());

            log.warn("收到下线规则的请求，成功下线了规则运算机: {}", ruleMetaBean.getRuleId());

        }

    }

}
